package com.yunlongn.async.service.impl;

import com.yunlongn.async.action.RetryActionWarp;
import com.yunlongn.async.function.RetryActionAdapter;
import com.yunlongn.async.function.RetryScheduleAction;
import com.yunlongn.async.function.RetryScheduleFail;
import com.yunlongn.async.service.RetrySchedulerService;
import com.yunlongn.async.util.spi.SpiLoader;

import java.util.ArrayList;
import java.util.List;
import java.util.ServiceLoader;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;

/**
 * @author yunlongn
 */
public class RetryScheduleServiceImpl implements RetrySchedulerService {

    /**
     * Action包装器
     */
    private static final List<RetryActionWarp> ACTION_WARPS;
    /**
     * 委托
     */
    private final RetrySchedulerService delegate;
    /**
     * 允许的最大并行Action数量
     */
    private final int maxAction;
    /**
     * 计算实时的并行Action数量
     */
    private final LongAdder COUNT = new LongAdder();

    static {
        // SPI获取Action包装器
        ACTION_WARPS = SpiLoader.loadAllServiceAndSort(Thread.currentThread().getContextClassLoader(),
                RetryActionWarp.class);

    }
    public RetryScheduleServiceImpl(ExecutorService executorService, int maxAction) {
        this.maxAction = maxAction;
        this.delegate = new SimpRetryScheduleServiceImpl(executorService);
    }

    @Override
    public <T> void retrySubmit(String retrySubmitId, RetryScheduleAction<T> action, int... periods) {
        this.retrySubmit(retrySubmitId, action, TimeUnit.SECONDS, periods);
    }

    @Override
    public <T> void retrySubmit(String retrySubmitId, RetryScheduleAction<T> retryScheduleAction, TimeUnit timeUnit, int... periods) {
        checkAllowSubmitNewAction();
        for (RetryActionWarp actionWarp : ACTION_WARPS) {
            // 包装执行器
            retryScheduleAction = actionWarp.warpAction(retrySubmitId, retryScheduleAction, timeUnit, periods);
        }
        COUNT.increment();

        retryScheduleAction = countWarp(retryScheduleAction);

        delegate.retrySubmit(retrySubmitId, retryScheduleAction, timeUnit, periods);
    }

    /**
     * 包装warp 执行完毕总数 -1
     */
    private <T> RetryScheduleAction<T> countWarp(RetryScheduleAction<T> retryScheduleAction) {
        return new RetryActionAdapter<T>(retryScheduleAction) {
            @Override
            public boolean execute(Object param) {
                boolean result = super.execute((T) param);
                COUNT.decrement();
                return result;
            }
        };
    }


    /**
     * 检查是否允许提交新的Action，根据maxAction判断
     */
    private void checkAllowSubmitNewAction() {
        if (maxAction < COUNT.intValue()) {
            throw new RuntimeException("已经超过配置所允许的最大并行Action数量");
        }
    }
}
